/**
 *
 */
package name.yzhu.wbdata;

import java.io.File;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;

import name.yzhu.cloud.hbase.HBaseUtil;
import name.yzhu.common.EnvVariable;
import name.yzhu.common.conf.PropertyConfig;
import name.yzhu.common.util.UUID;
import name.yzhu.httpserver.AdminServlet;
import name.yzhu.httpserver.HTTPServletServer;
import name.yzhu.httpserver.ServerException;
import name.yzhu.wbdata.db.DBUtil;
import name.yzhu.wbdata.server.HBaseServelt;
import name.yzhu.wbdata.server.TokenServelt;
import name.yzhu.wbdata.server.WBServerManager;
import name.yzhu.wbdata.weibo.WeiboHbaseService;
import name.yzhu.wbdata.weibo.WeiboService;

/**
 * @author yhzh
 *
 */
public final class WBDataUtil{
	static private final Log LOG = LogFactory.getLog(WBDataUtil.class);

	private static final String PERSIST_TO_HBASE = "persist_to_hbase";

	private static PropertyConfig proCFG = new PropertyConfig();
	private static HTTPServletServer wbServer = null;
	private static WBServerManager wbsm = new WBServerManager();
	private static WeiboService dataServer = null;

	private static Configuration hbaseCFG = null;

	public static void initialSystem() throws Exception {
		EnvVariable.initialGlobalEnvVariable();

		initialConfig();

		if (proCFG.getBoolean(PERSIST_TO_HBASE))
		    initalizeHBase();
		
		initialServer() ;

		DBUtil.initialDatabase(proCFG);

		dataServer.init();

		Runtime.getRuntime().addShutdownHook(new Thread() {

			@Override
			public void run() {
				LOG.info("shutdown hook is running...");
				if (!wbsm.isStopped())
				    WBDataUtil.destroy();
			}});
	}

	public static void run()  {
		try {
			wbServer.start();
			dataServer.start();

			while (!wbsm.isStopped()) {
				Thread.sleep(500);
			}
			destroy();
		} catch (ServerException e) {
			LOG.error("ServerException", e);
		} catch (InterruptedException e) {
			LOG.error("InterruptedException", e);
		}
	}

	public static void destroy() {
		LOG.info("destroy begin...");
		try {
			wbServer.stop();
			dataServer.stop();
			wbServer.join();
			dataServer.join();
		} catch (ServerException e) {
			LOG.error("ServerException", e);
		}
		LOG.info("destroy end...");
	}

	public static void initialConfig() throws IOException  {
		proCFG.load(DBUtil.DEFAULT_SYSTEM_CFG_FILE);
	}
	
	public static void initialServer() {
		AdminServlet as = new AdminServlet();
		as.setStoppable(wbsm);
		wbServer = new HTTPServletServer(as);
		wbServer.addServlet(as);
		wbServer.addServlet(new TokenServelt());
		
		if (proCFG.getBoolean(PERSIST_TO_HBASE))
		    wbServer.addServlet(new HBaseServelt(hbaseCFG));
		
		wbServer.setWebBase(EnvVariable.getWorkingDirectory()+File.separator+Constant.WEB_DIR_NAME);
		wbServer.setWebRootpath(Constant.WEB_ROOT_PATH);

		if (proCFG.getBoolean(PERSIST_TO_HBASE))
			dataServer = new WeiboHbaseService(wbsm, proCFG, hbaseCFG);
		else 
		    dataServer  = new WeiboService(wbsm, proCFG);
	}

	public static void initalizeHBase() throws IOException {
		String zkQ = proCFG.getString(name.yzhu.cloud.Constant.ZOOKEEPER_QUORUM);
		if (zkQ == null) 
			throw new IOException("Please specify "+name.yzhu.cloud.Constant.ZOOKEEPER_QUORUM);
		Map<String, Object> map = new HashMap<String, Object>();
		map.put(name.yzhu.cloud.Constant.ZOOKEEPER_QUORUM, zkQ);
		
		hbaseCFG  = HBaseUtil.getHBaseConfig(map.entrySet());
	}
	
	public WBServerManager getWbsm() {
		return wbsm;
	}

	public static  PropertyConfig getProCFG() {
		return proCFG;
	}

	public static String getUUID(String ds, String id) {
		return UUID.generateUUID(ds, id);
	}

	public static Configuration getHbaseCFG() {
		return hbaseCFG;
	}
	
}
